第十五章:Python Flask 後端開發基礎
本章節將帶您從零開始建置 Python Flask 後端開發環境,並建立您的第一支 API 程式。
步驟一:基礎環境準備
- 安裝 Python:前往官網下載最新版安裝檔。⚠️ 致命防呆:安裝時務必勾選
Add Python to PATH,否則後續在終端機輸入 Python 指令時會出現找不到指令的錯誤。 - 安裝 Postman:下載並安裝 Postman,這是一款強大的 API 測試工具,負責模擬前端發送請求給後端,是後續開發與測試必定會用到的神器。
步驟二:VS Code 開發套件配置
開啟 VS Code,並在左側延伸模組 (Extensions) 中搜尋並安裝以下必備套件,打造完美的開發環境:
- Python:提供 Python 語言的核心支援與環境偵測。
- Pylance:微軟官方的高效能 Python 語法提示、自動完成與錯誤檢查工具。
- Flask Snippets:提供 Flask 常用的程式碼片段,輸入幾個字就能產生完整架構,大幅加速開發。
- SQLite:用於後續在 VS Code 內直接檢視與管理 SQLite 資料庫的內容。
步驟三:建立虛擬環境與安裝 Flask
在 VS Code 中開啟您的專案資料夾,並開啟終端機 (Terminal),依序執行以下指令:
# 1. 建立虛擬環境 (Virtual Environment)
python -m venv venv
# 2. 啟動虛擬環境 (Windows 系統適用)
venv\Scripts\activate
# 3. 安裝 Flask 框架套件
pip install flask
虛擬環境能為每個專案建立獨立的 Python 執行空間,確保不同專案之間的套件版本不會互相干擾。啟動成功後,終端機的命令提示字元前面會多出一個綠色的
(venv)
標示,代表您已進入該專案的專屬環境。
步驟四:撰寫第一支 Flask 程式 (app.py)
在專案根目錄下建立一個名為 app.py 的檔案,並輸入以下測試程式碼:
from flask import Flask
app = Flask(__name__)
@app.route('/')
def home():
return 'Hello Flask!'
if __name__ == '__main__':
app.run(debug=True)
程式碼原理解析:
這短短幾行程式碼就架起了一個微型伺服器,我們逐行解析其背後的運作邏輯:
app = Flask(__name__):建立伺服器物件。這是在實體化 Flask 應用程式。@app.route('/'):設定網址路徑的對應行為(路由)。'/'代表網站的最根部(首頁)。def home()::當有人訪問/這個路徑時,就會觸發並執行這個自訂函式。return 'Hello Flask!':這是 API 的回應內容。當前端發送 GET 請求過來時,就會在畫面上看到這句回傳的字串。app.run(debug=True):啟動伺服器。預設網址是http://127.0.0.1:5000。括號內加上debug=True是開發時的實用小技巧,它可以讓伺服器在您一存檔程式碼時,自動重新載入套用新邏輯,不需反覆手動重啟。
步驟五:啟動伺服器
在終端機 (請確保 (venv) 虛擬環境仍在啟動狀態) 輸入以下指令來運行這支程式:
python app.py
執行成功後,終端機會顯示伺服器正在運行的提示。請打開您的網頁瀏覽器,在網址列輸入 http://127.0.0.1:5000,您就會看到乾淨的白底黑字顯示著「Hello
Flask!」,恭喜您成功邁入後端開發的世界!
步驟六:建立第一個 POST API (新增產品)
在了解基本的 Flask 伺服器運作後,我們來實作一個真實的後端功能:接收前端傳來的資料並存入資料庫。這裡我們以「新增產品」的 POST API 為例。
A. RESTful API 路由規劃
在設計 CRUD (新增、讀取、更新、刪除) 功能時,通常會遵循 API 的設計規範,為不同操作搭配適合的 HTTP 方法:
- 取得全部商品 (R):
GET /products(成功回傳狀態碼 200 OK) - 新增商品 (C):
POST /add_product(成功回傳狀態碼 201 Created) - 更新商品 (U):
PUT /products/<id>(成功回傳狀態碼 200 OK) - 刪除商品 (D):
DELETE /products/<id>(成功回傳狀態碼 200 OK)
B. 撰寫「新增產品」程式碼
請將您的 app.py 改寫為以下內容。為了方便演示,我們暫時使用一個 Python 的 list 陣列來模擬資料庫存取:
from flask import Flask, request, jsonify
app = Flask(__name__)
# 建立一個模擬的產品資料庫 (用 list 暫存)
products = []
# 建立 POST API 路由
@app.route('/add_product', methods=['POST'])
def add_product():
# 1. 接收與解析 JSON 格式資料
try:
# force=False 不強制解析,若格式錯誤會觸發例外
data = request.get_json(force=False)
if data is None:
raise ValueError
except Exception:
# 發生例外時,回傳自訂錯誤訊息與 400 狀態碼
return jsonify({"error": "請傳送正確的 JSON 格式資料"}), 400
# 2. 從資料中取得產品名稱與價格
name = data.get('name')
price = data.get('price')
# 3. 簡單的防呆驗證:若缺資料則回傳錯誤
if not name or not price:
return jsonify({"error": "請提供產品名稱和價格"}), 400
# 4. 建立產品字典,加上自動遞增的 ID
product = {
"id": len(products) + 1,
"name": name,
"price": price
}
# 5. 將產品加入模擬資料庫
products.append(product)
# 6. 回傳新增成功與產品資訊 (HTTP 狀態碼 201)
return jsonify({"message": "產品新增成功", "product": product}), 201
if __name__ == '__main__':
# 注意:POST 方法一定要用工具 (如 Postman) 來發送請求測試
app.run(debug=True)
核心觀念解析:
request.get_json(force=False): 用來解析前端傳來的資料。設定force=False代表不強制解析,如果對方傳來的格式錯誤,就會回傳None或觸發例外。- 例外處理
try-except: 這是後端必備的防護網。它可以處理非 JSON 格式傳入的情況,並回應適當錯誤訊息,避免因為一個錯字導致整個伺服器崩潰。 jsonify(): 這是 Flask 模組內建的 function。功能是將 Python 的字典格式轉換為標準的 JSON 格式,讓前端或 Postman 看得懂回傳內容。- HTTP 狀態碼: 在
return語法中可直接設定狀態碼,例如400代表前端請求有誤,201則是專門用來表示「新增成功 (Created)」的標準代碼。
C. 使用 Postman 測試 API
因為這是一個 POST 請求,我們無法直接在瀏覽器網址列(預設為 GET)進行測試,必須使用 Postman 傳送 JSON 資料給伺服器:
- 打開 Postman,將請求方法切換為 POST。
- 輸入 API 網址:
http://127.0.0.1:5000/add_product。 - 在下方的頁籤選擇 Body ➔ raw ➔ 格式下拉選單選擇 JSON。
- 輸入測試資料:
{"name": "蘋果", "price": 30},點擊 Send 送出。伺服器應成功回應「產品新增成功」與對應內容。
寫完 API 後,請嘗試用各種「錯誤」的方式來攻擊自己的程式,驗證防呆機制:
- 格式錯誤: 傳送純文字、HTML 或是漏掉大括號,觀察伺服器是否會如期回應「請傳送正確的 JSON 格式資料」。
- 漏傳欄位: 不傳
price或name,測試是否會收到「請提供產品名稱和價格」的 400 錯誤。 - 多餘欄位:
自行設計一個商品欄位並嘗試加進去(例:
{"name":"可樂","price":25, "category":"飲料"}),觀察未被處理的欄位是否會被安全地擋下或忽略。 - 用錯方法 (Method Not Allowed): 嘗試使用
GET去打這支 POST API,觀察出現的錯誤訊息。未來遇到此錯誤,就代表你在 Postman 選錯方法了。 - 錯誤的 Content-Type: 在 Postman 選擇
x-www-form-urlencoded,傳送name=茶&price=40,觀察是否能被擋下。 - 功能驗證: 嘗試連續新增多筆產品,觀察
id欄位是否有按照預期自動累加。
步驟七:建立 GET API (查詢所有產品)
除了新增資料,我們也需要一個 GET API 來讓前端讀取完整的產品列表。
# 1. 建立一個模擬的產品資料庫 (用 list 暫存)
products = [
{"id": 1, "name": "綠茶", "price": 35},
{"id": 2, "name": "紅茶", "price": 30}
]
# 2. 建立 /products 路由,使用 GET 方法查詢所有產品
@app.route('/products', methods=['GET'])
def get_all_products():
# 回傳所有產品資料與 200 成功狀態碼
return jsonify({"products": products}), 200
步驟八:API 資料驗證與錯誤處理技巧
為了確保後端系統的穩定性,當接收前端傳來的 POST 或 PUT 資料時,必須進行嚴格的防呆驗證。以下是四種必備的驗證技巧:
# 1. 必填驗證:若缺資料則回傳錯誤
if not name or not price:
return jsonify({"error": "請提供產品名稱和價格"}), 400
# 2. 型別驗證:使用 isinstance() 確認型別,名稱需為字串,價格需為數字
if not isinstance(name, str):
return jsonify({"error": "產品名稱必須是文字格式"}), 400
if not isinstance(price, (int, float)):
return jsonify({"error": "產品價格必須是數字格式"}), 400
# 3. 邏輯驗證:價格不能為負數
if price < 0:
return jsonify({"error": "產品價格不能為負數"}), 400
# 4. 重複性驗證:防止名稱重複或資料污染
for p in products:
if p['name'] == name:
return jsonify({"error": "產品名稱已存在,請重新命名"}), 400
一旦驗證不通過,我們一律回傳
400 狀態碼,明確告知前端這是「用戶端請求錯誤」,並附帶具體的 error 提示訊息,幫助前端除錯。
附錄:後端常用 HTTP 狀態碼總覽
在設計 RESTful API 時,正確的狀態碼能讓前後端溝通更順暢:
| 狀態碼 | 類別 | 用途說明 |
|---|---|---|
| 200 OK | 成功 | 一般成功 (常用於 GET 查詢、PUT 更新、DELETE 刪除) |
| 201 Created | 成功 | 新增成功 (常用於 POST 建立新資料) |
| 204 No Content | 成功 | 刪除成功,但無內容回傳 |
| 400 Bad Request | 錯誤 | 用戶端請求錯誤 (例如:資料格式錯、缺必填欄位) |
| 401 Unauthorized | 錯誤 | 未登入 / 授權失敗 |
| 403 Forbidden | ▲ 錯誤 | 權限不足 (已登入,但無權限操作) |
| 404 Not Found | ▲ 錯誤 | 找不到資源 (API 網址打錯或查無此 ID) |
| 500 Server Error | 錯誤 | 伺服器端錯誤 (後端程式寫錯導致當機) |
步驟九:更新與刪除產品資料 (PUT & DELETE)
在實作更新與刪除 API 之前,我們會頻繁地透過 ID 尋找產品。因此,先建立一個輔助函式 find_product 來簡化後續的程式碼。
# 輔助函式:用 ID 找產品,找不到則回傳 None
def find_product(pid: int):
for p in products:
if p["id"] == pid:
return p
return None
A. 建立更新產品資料的 PUT API
更新功能需支援「部分更新」(例如只改價格不改名字),且網址需使用動態路由來接收前端傳來的產品 ID。
@app.route("/products/<int:pid>", methods=["PUT"])
def update_product(pid):
# 1. 解析資料 (省略部分 try-except 基礎驗證)
data = request.get_json(force=False)
name = data.get("name", None)
price = data.get("price", None)
# 2. 驗證:至少提供一個欄位
if name is None and price is None:
return jsonify({"error": "至少需提供 name 或 price 之一"}), 400
# 3. 尋找目標產品
target = find_product(pid)
if target is None:
return jsonify({"error": "找不到此產品"}), 404
# 4. 若要改名,檢查是否與「其他產品」重複
if name is not None:
for p in products:
if p["id"] != pid and p["name"] == name:
return jsonify({"error": "品名已存在不可以重複"}), 409
target["name"] = name # 執行更新
# 5. 更新價格
if price is not None:
target["price"] = price
return jsonify({"message": "產品更新成功", "product": target}), 200
B. 建立刪除產品資料的 DELETE API
刪除功能的邏輯相對簡單:確認該 ID 存在後,直接從串列中移除即可。
@app.route("/products/<int:pid>", methods=["DELETE"])
def delete_product(pid):
# 1. 尋找目標產品
target = find_product(pid)
if target is None:
return jsonify({"error": "找不到此產品"}), 404
# 2. 執行刪除
products.remove(target)
return jsonify({
"message": "刪除成功",
"deleted": target,
"count": len(products)
}), 200
完成 API 後,請針對以下情境進行測試以確保系統強健度:
- 動態路由測試: 請求網址應為
http://127.0.0.1:5000/products/1(結尾的 1 即為產品 ID)。 - 部分更新測試 (PUT): 嘗試在 Body 中只傳送
{"price": 50},觀察名稱是否保留原狀,且價格成功更新。 - 錯誤防呆測試:
- 修改為負數價格,測試是否被擋下 (400 錯誤)。
- 將產品改名為資料庫中已存在的名稱,觀察是否正確觸發 HTTP 409 (Conflict) 狀態碼。
- 對不存在的 ID (例如
/products/999) 執行 PUT 或 DELETE,確認回傳 404 (Not Found) 錯誤。
步驟十:導入真實資料庫 (SQLite)
前面的章節我們使用 Python 的 list 暫存資料,但伺服器一重啟資料就會消失。現在我們正式導入輕量級的關聯式資料庫 SQLite
來達成資料的永久保存。
A. 初始化資料庫與資料表
在伺服器啟動前,我們需要一個 init_db() 函式來確保資料庫檔案與資料表已經建立好:
import sqlite3
# 啟動時自動建立資料庫與資料表 (如果尚未存在)
def init_db():
conn = sqlite3.connect('products.db')
cursor = conn.cursor()
# 建立 products 資料表
cursor.execute('''
CREATE TABLE IF NOT EXISTS products (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL,
price REAL NOT NULL
)
''')
conn.commit()
conn.close()
if __name__ == '__main__':
init_db() # 啟動時自動初始化資料庫
app.run(debug=True)
B. 改寫新增產品 API (POST)
將原本寫入 list 的邏輯,改為使用 SQL 指令寫入資料庫:
# 1. 連線資料庫
conn = sqlite3.connect("products.db")
cursor = conn.cursor()
# 2. 檢查是否已存在同名商品 (避免重複)
cursor.execute("SELECT id FROM products WHERE name=?", (name,))
if cursor.fetchone():
conn.close()
return jsonify({"error": "此產品名稱已存在"}), 409
# 3. 執行 SQL 寫入資料庫
cursor.execute("INSERT INTO products (name, price) VALUES (?,?)", (name, price))
conn.commit()
# 4. 取得剛剛新增的自動遞增 ID
product_id = cursor.lastrowid
conn.close()
C. 改寫查詢產品 API (GET)
從 SQLite 取出的資料是 Tuple (例如 (1, "紅茶", 30.0)),我們必須將它轉換為前端看得懂的 JSON (字典) 格式。
conn = sqlite3.connect('products.db')
cursor = conn.cursor()
cursor.execute("SELECT id, name, price FROM products")
rows = cursor.fetchall() # 取出所有資料
conn.close()
products = []
# 將每筆 row (tuple) 轉換成字典,組成陣列
for row in rows:
products.append({
"id": row[0],
"name": row[1],
"price": row[2]
})
return jsonify({"products": products}), 200
D. 改寫更新與刪除 API (PUT & DELETE)
導入 SQLite 後,更新與刪除資料不再是操作 Python 的 List,而是要撰寫對應的 SQL 指令 (UPDATE 與
DELETE)。網址同樣使用動態路由來接收 product_id。
1. 更新產品資料 (PUT)
@app.route('/update_product/<int:product_id>', methods=['PUT'])
def update_product(product_id):
try:
data = request.get_json(force=False)
name = data.get('name')
price = data.get('price')
# (省略基礎的格式與空值驗證...)
# 連線並執行 SQL 更新指令
conn = sqlite3.connect('products.db')
cursor = conn.cursor()
cursor.execute("UPDATE products SET name=?, price=? WHERE id=?", (name, price, product_id))
conn.commit()
conn.close()
return jsonify({"message": f"產品編號 {product_id} 已更新"}), 200
except Exception as e:
return jsonify({"error": str(e)}), 500
2. 刪除產品資料 (DELETE) 與防呆技巧
刪除資料時有一個常見陷阱:如果前端傳了一個**不存在的 ID**,SQL 執行 DELETE 也不會報錯。我們必須利用 cursor.rowcount
來判斷到底有沒有資料被刪掉。
@app.route('/delete_product/<int:product_id>', methods=['DELETE'])
def delete_product(product_id):
try:
conn = sqlite3.connect('products.db')
cursor = conn.cursor()
# 執行 SQL 刪除指令
cursor.execute("DELETE FROM products WHERE id=?", (product_id,))
conn.commit()
# 🎯 關鍵:取得剛才執行的指令「影響了幾筆資料」
deleted_count = cursor.rowcount
conn.close()
# 若影響筆數為 0,代表資料庫裡根本沒這個產品
if deleted_count == 0:
return jsonify({"error": f"找不到ID為 {product_id} 的產品,無法刪除"}), 404
return jsonify({"message": f"產品編號 {product_id} 已刪除"}), 200
except Exception as e:
return jsonify({"error": str(e)}), 500
在使用 SQLite 進行
UPDATE 或 DELETE 時,只要 SQL 語法沒寫錯,即使找不到目標 ID,執行也不會發生
Exception。因此務必養成習慣,利用 cursor.rowcount 檢查影響筆數,若為 0 則手動回傳
404 Not Found 錯誤,這樣前端才能正確提示使用者「資料不存在,不可以被刪除」。
步驟十一:補充工具 - Headers 權限驗證與清空資料
有時候我們需要一個具有殺傷力的 API (如清空所有資料),這種 API 絕對不能讓人隨便呼叫,必須加上權限驗證 (Authorization)。我們可以透過檢查請求的 Headers 來實現。
@app.route('/clear_products', methods=['POST'])
def clear_products():
# 1. 從 headers 讀取自訂的 Token (X-Admin-Token)
auth = request.headers.get('X-Admin-Token')
# 2. 驗證密碼是否正確
if auth != 'my_secret_password_1234567':
return jsonify({"error": "無權限,請提供有效密碼"}), 403
# 3. 若通過驗證,執行刪除
try:
conn = sqlite3.connect('products.db')
cursor = conn.cursor()
cursor.execute("DELETE FROM products") # 刪除所有資料
conn.commit()
conn.close()
return jsonify({"message": "所有產品資料已清除"}), 200
except Exception as e:
return jsonify({"error": f"清除失敗: {str(e)}"}), 500
要測試此 API,不能只把資料放在 Body。您必須在 Postman 的 Headers 頁籤中,手動新增一行:
Key: X-Admin-TokenValue: my_secret_password_1234567
403 FORBIDDEN 錯誤。
專案三:建立 MySQL 會員 API 系統
在這個進階專案中,我們將導入正式的 MySQL 資料庫,並透過 Flask 實作一套具備安全性的會員註冊、登入與權限管理機制。以下是專案的第一步驟:確認功能與環境建置。
步驟一:功能規劃與環境準備
1. API 功能簡介
我們預計實作四支核心 API,涵蓋完整的會員狀態生命週期:
- 會員註冊:
POST /api/register(接收帳號密碼並建立新使用者) - 會員登入:
POST /api/login(驗證成功後,核發並回傳一組 auth_token) - 取得當前登入者:
GET /api/me(透過 token 獲取目前登入者的詳細資訊) - 管理員專屬功能:
GET /api/admin/users(列出全部會員,僅限具有 admin 權限的 token 呼叫)
2. 系統環境與套件需求
在 VS Code 中,建議安裝 Pylance 與 Black Formatter 延伸模組來提升開發體驗與程式碼排版品質。接著,請確認已安裝以下關鍵 Python 套件:
flask:建構後端伺服器核心。flask-cors:處理跨來源資源共用 (CORS),允許前端跨網域呼叫這台伺服器的 API。pymysql:讓 Python 能夠連線並操作 MySQL 資料庫的驅動套件。werkzeug:(隨 Flask 一起安裝) 我們將使用其內建的security模組,來進行密碼的單向雜湊加密。
3. MySQL 資料庫設計 (users 資料表)
請先在您的 MySQL 資料庫(可透過 phpMyAdmin 管理)中建立一個名為 testdb 的資料庫,並建立一張 users
資料表。為了安全性與狀態管理,必須包含以下關鍵欄位:
- id: 整數 (int),設定為主鍵 (Primary Key) 並勾選自動遞增 (AUTO_INCREMENT)。
- username: 帳號字串 (varchar),必須設定為唯一值 (Unique Index),避免重複註冊。
- password_hash: 密碼雜湊字串 (varchar)。基於資安考量,絕對不可以儲存明碼,此欄位專門用來存放加密轉換後的亂碼結果。
- level: 會員等級 (varchar),用於權限控管的判斷,預設值為
normal(一般會員),特定人員可改為admin(管理員)。 - auth_token: 授權字串 (varchar)。每次登入成功後,由系統隨機產生的一組 Token,用來代表該使用者的登入狀態。
- created_at: 註冊時間 (timestamp),可設定預設值為
CURRENT_TIMESTAMP,由資料庫自動記錄寫入時間。
步驟二:app.py 基本架構與資料庫連線
在開始撰寫會員 API 之前,請先確保在 VS Code 的終端機 (需處於虛擬環境中) 安裝必要的外部套件:
pip install flask flask-cors pymysql
接著,在 app.py 中匯入所需套件,設定全域變數,並建立 MySQL 的連線函式 get_connection():
from flask import Flask, request, jsonify
from flask_cors import CORS
import pymysql
from werkzeug.security import generate_password_hash, check_password_hash
import secrets
app = Flask(__name__)
# 啟用 CORS,讓「不同網址的前端」可以呼叫這個 API (不影響 Postman 測試)
CORS(app)
# === MySQL 連線設定 ===
DB_HOST = "localhost"
DB_USER = "owner" # 請替換為您的資料庫帳號
DB_PASSWORD = "123456" # 請替換為您的資料庫密碼
DB_NAME = "testdb" # 目標資料庫名稱
# 建立 MySQL 連線的共用函式
def get_connection():
return pymysql.connect(
host=DB_HOST,
user=DB_USER,
password=DB_PASSWORD,
database=DB_NAME,
charset="utf8mb4",
# 關鍵設定:讓資料庫查詢結果自動轉為 Python 字典 (Dictionary) 格式
cursorclass=pymysql.cursors.DictCursor,
)
# 測試用 API:確認伺服器是否正常運作
@app.route("/api/ping")
def ping():
return jsonify({"message": "pong"})
# --- 預留會員相關 API 區塊 ---
# 1. 會員註冊 /api/register
# 2. 會員登入 /api/login
# 3. 取得目前登入者資訊 /api/me
# 4. 管理員專用 API /api/admin/users
# 啟動伺服器的程式入口
if __name__ == "__main__":
app.run(debug=True)
flask_cors.CORS: 解決「跨來源資源共用 (CORS)」的問題。因為前端網頁和後端 API 經常運行在不同的 Port 或網域,若未開啟此功能,瀏覽器基於安全性會阻擋前端發送的 AJAX 請求。pymysql.cursors.DictCursor: 預設情況下,資料庫查詢回傳的是 Tuple (例如:(1, 'test', '...'))。加上此參數後,資料庫回傳的結果會變成帶有欄位名稱的字典 (例如:{'id': 1, 'username': 'test'}),這讓後續資料處理變得非常直觀且不易出錯。werkzeug.security與secrets: 我們在此階段先將這兩個模組匯入,它們將在下一步負責「密碼雜湊加密」與「產生安全隨機 Token」的關鍵資安任務。
步驟三:會員註冊 API (POST /api/register)
這支 API 負責處理新會員的註冊請求。基於資安考量,我們必須在資料寫入資料庫前,完成防呆檢查與密碼加密的動作。以下為完整的實作與測試步驟:
1. 接收前端傳來的 JSON
首先從請求中讀取資料,並確保必要欄位不為空。
- 讀取
username和password。 - 檢查是否缺少欄位。
@app.route("/api/register", methods=["POST"])
def register():
# 讀取 username 和 password
data = request.get_json()
username = data.get("username")
password = data.get("password")
# 檢查是否缺少欄位
if not username or not password:
return jsonify({"error": "缺少 username 或 password"}), 400
conn = get_connection()
try:
with conn.cursor() as cursor:
2. 檢查帳號是否已存在
在寫入前,必須確保資料庫中沒有重複的帳號。
- 執行
SELECT id FROM users WHERE username = %s。 - 如果查到資料 → 回傳錯誤:
{"error": "帳號已存在"}。
# 檢查帳號是否已存在
cursor.execute("SELECT id FROM users WHERE username=%s", (username,))
exist = cursor.fetchone()
if exist:
return jsonify({"error": "帳號已存在"}), 400
3. 產生密碼雜湊
這是會員系統最重要的資安防護。
- 使用
generate_password_hash(password)進行加密轉換。 - 永遠不要把原始密碼存進資料庫。
# 產生密碼雜湊 (需先 pip install cryptography)
password_hash = generate_password_hash(password)
4. 寫入資料庫
將驗證過且加密的資料正式寫入資料表。
- 執行
INSERT INTO users (username, password_hash, level) VALUES (..., "normal")。 - 統一給新會員預設等級
normal。
# 新增使用者並寫入資料庫
cursor.execute(
"INSERT INTO users (username, password_hash, level) VALUES (%s, %s, %s)",
(username, password_hash, "normal")
)
conn.commit() # 確認寫入
5. 回傳註冊成功訊息
結束資料庫操作,回傳成功狀態給前端。
- 回傳 JSON:
{"message": "register ok"}。
# 回傳註冊成功訊息
return jsonify({"message": "register ok"}), 200
finally:
conn.close() # 確保資料庫連線關閉
完成
app.py 的撰寫後,請進行以下測試以確保流程暢通:
- 開啟
app.py: 在終端機執行程式以啟動 Flask 伺服器。 - Postman 測試: 測試使用 Postman 傳
POST請求與新帳號的 JSON 資料至/api/register。 - 資料庫檢視: 開啟 MySQL (例如 phpMyAdmin) 檢視
users資料表是否有新帳號被建立,如果成功建立即已經打通這段。
步驟四:會員登入 API (POST /api/login)
這支 API 負責驗證使用者身分,並在驗證成功後核發一組 auth_token 作為後續操作的憑證。以下是完整的實作步驟:
1. 接收帳號與密碼
同樣從請求的 JSON 本體中讀取資料,並執行基礎欄位檢查。
@app.route("/api/login", methods=["POST"])
def login():
data = request.get_json()
username = data.get("username")
password = data.get("password")
if not username or not password:
return jsonify({"error": "請提供帳號和密碼"}), 400
2. 查詢資料庫是否有該帳號
連線至資料庫,搜尋該 username 是否存在於 users 資料表中。
- 執行 SQL:
SELECT id, username, password_hash, level FROM users WHERE username = %s。 - 若查無此人 → 回傳 401 錯誤:
{"error": "帳號或密碼錯誤"}。
3. 比對密碼
使用加密工具比對前端傳來的明碼與資料庫中的雜湊值是否吻合。
- 使用
check_password_hash(user["password_hash"], password)進行驗證。 - 比對失敗 → 回傳 401 錯誤:
{"error": "帳號或密碼錯誤"}。 - 💡 資安細節:不論是帳號錯還是密碼錯,一律回傳相同的錯誤訊息,以防駭客透過錯誤提示猜測帳號是否存在。
4. 產生 Token
當身分驗證通過,系統會產生一組唯一的隨機字串作為該次登入的「通行證」。
- 使用
secrets.token_hex(16)產生 32 字元的隨機 Token。 - 這組 Token 在後續請求中代表該使用者的「登入狀態」。
5. 把 Token 更新回資料庫
將產生的 Token 存入資料庫,以便未來驗證其他 API 的呼叫權限。
- 執行 SQL:
UPDATE users SET auth_token = %s WHERE id = %s。
6. 回傳登入成功資訊與 Token
將結果打包成 JSON 回傳給前端儲存。
# 完整的登入成功回傳邏輯
return jsonify({
"message": "login ok",
"token": token,
"username": user["username"],
"level": user["level"]
}), 200
使用 Postman 進行
POST 測試:
- 輸入網址
http://127.0.0.1:5000/api/login並在 Body 帶入 JSON 資料。 - 點擊 Send 後,觀察是否獲得
"message": "login ok"以及一串長長的token。 - 資料庫檢查:確認 MySQL 中該名使用者的
auth_token欄位是否已從 NULL 變更為剛產生的隨機字串。
步驟五:取得目前登入者資訊 (GET /api/me)
這支 API 的核心用途是查詢:「現在攜帶這個 Token 來發送請求的是哪一位會員?」這在前端開發中非常重要,通常用於網頁重新載入時,自動恢復使用者的登入狀態與畫面。
API 邏輯流程
我們將驗證 Token 的複雜邏輯抽離成一個獨立的工具函式 get_current_user_from_request()(將在下一階段詳細實作),讓 API
本體的程式碼保持簡潔:
- 呼叫身分驗證工具: 嘗試從本次 Request 中解析出使用者身分。
- 攔截未授權存取: 如果回傳
None,代表該請求「沒帶 Token」或「Token 已失效/造假」,系統將無情拒絕並回傳401 Unauthorized錯誤:{"error": "未登入或token無效"}。 - 放行並回傳資料: 若成功解析出合法使用者,則回傳該名會員的基本資訊(如 id、帳號、權限等級)。
@app.route("/api/me", methods=["GET"])
def me():
# 1. 解析 Request 取得當前使用者
user = get_current_user_from_request()
# 2. 判斷是否為合法登入狀態
if not user:
return jsonify({"error": "未登入或token無效"}), 401
# 3. 回傳會員簡單資訊
return jsonify({
"id": user["id"],
"username": user["username"],
"level": user["level"]
}), 200
這支 API 是「認 Token 不認人」的,因此測試時不能只打網址,必須手動將剛剛登入拿到的 Token 放進 HTTP Request Headers 中:
- 在 Postman 設定 HTTP 請求方法為
GET,網址輸入http://127.0.0.1:5000/api/me。 - 切換到下方的 Headers 頁籤。
- 新增一組 Key-Value 對應:
- Key: 輸入
Authorization - Value: 輸入
Bearer 您的Token字串(注意:Bearer單字後面必須空一格,再貼上亂碼 Token)。
- Key: 輸入
- 點擊 Send 送出,若設定正確,伺服器就會認出這個 Token,並回傳對應的會員 JSON 資訊。
步驟六:實作登入狀態驗證工具函式 (Helper Functions)
為了保持 API 路由程式碼的簡潔,我們將「解析 Header」與「查詢資料庫」這兩項繁瑣的驗證動作,封裝成獨立的工具函式。
1. 工具函式 (一):get_user_by_token(token)
此函式的職責很單純:給它一組 Token 字串,它就去資料庫把這名會員找出來。
def get_user_by_token(token):
# 如果沒傳 token 進來,直接中斷回傳 None
if not token:
return None
conn = get_connection()
try:
with conn.cursor() as cursor:
# 找出 auth_token 欄位符合的該筆 user
cursor.execute(
"SELECT id, username, level FROM users WHERE auth_token = %s",
(token,)
)
user = cursor.fetchone()
return user # 回傳使用者資訊(字典),若找不到則自動回傳 None
finally:
conn.close()
2. 工具函式 (二):get_current_user_from_request()
這是我們在 API 中實際呼叫的函式,它負責去 HTTP Header 中「挖出」前端夾帶的 Token,然後交給前面的函式去查驗。
- 從 HTTP Header 裡讀取
Authorization欄位。 - 檢查前綴是否為
Bearer,若是,則將真實的 Token 切割出來。
def get_current_user_from_request():
# 1. 抓取 Authorization 欄位,若無則預設為空字串
auth_header = request.headers.get("Authorization", "")
# 2. 預期格式為: "Bearer "
if auth_header.startswith("Bearer "):
# 透過空白字元分割,取得索引值 [1] 的 Token 本體
token = auth_header.split(" ", 1)[1]
else:
token = None
# 3. 丟給 get_user_by_token() 查資料庫
user = get_user_by_token(token)
return user
在撰寫程式碼時,您會發現我們使用
.split(" ", 1)[1] 來切掉 Bearer 這個單字。這是因為在 OAuth 2.0 與現代 API
規範中,Bearer (持票人) 是一種標準的授權類型聲明。它告訴伺服器:「攜帶這張『票』(Token)
的人,就擁有對應的權限」。這種標準化的前綴設計,可以讓後端更容易區分不同類型的授權機制。
步驟七:管理員專用 API (GET /api/admin/users)
這支 API 負責列出系統內所有的會員資料。為了保護敏感資訊,我們實作了「兩層權限控制」機制,確保只有真正的管理員才能存取此功能。
權限控制與資料庫讀取流程
- 第一層:有沒有登入? 透過
get_current_user_from_request()檢查 Token 是否有效。若沒帶 Token 或無效,直接回傳 401 錯誤:{"error": "未登入或token無效"}。 - 第二層:是不是管理員 (admin)? 檢查解析出來的
current_user["level"]是否等於"admin"。若是一般會員,回傳 403 錯誤:{"error": "沒有權限,只有admin 可以使用這個功能"}。 - 放行並讀取資料: 若以上兩關皆合法通過,則執行 SQL 指令
SELECT id, username, level, created_at FROM users ORDER BY id,將所有會員列表打包回傳。
@app.route("/api/admin/users", methods=["GET"])
def admin_get_all_users():
# 1. 先確認有沒有登入 (token 合法)
current_user = get_current_user_from_request()
if not current_user:
# 沒登入或 token 無效
return jsonify({"error": "未登入或token無效"}), 401
# 2. 確認是不是 admin
if current_user["level"] != "admin":
# 有登入,但不是管理員
return jsonify({"error": "沒有權限,只有 admin 可以使用這個功能"}), 403
# 3. 真正執行「列出全部會員」的動作
conn = get_connection()
try:
with conn.cursor() as cursor:
cursor.execute(
"SELECT id, username, level, created_at FROM users ORDER BY id"
)
users = cursor.fetchall()
# 4. 回傳全部會員資料
return jsonify({"users": users})
finally:
conn.close()
在這個範例中,我們可以很清楚地實踐
401 Unauthorized 與 403 Forbidden 的運用差異:
- 401 Unauthorized: 代表「你是誰?我認不出你」。通常是因為沒帶 Token、Token 過期或造假。
- 403 Forbidden: 代表「我知道你是誰,但你不能進來」。通常是成功登入了 (Token 有效),但你的權限等級 (level) 不足,被伺服器拒絕存取。
level 欄位改成 admin。接著在 Postman
分別使用「一般帳號的 Token」與「管理員帳號的 Token」放入 Headers 來發送請求,觀察伺服器是否會精準地擋下一般帳號 (回傳 403) 並放行管理員 (回傳 200 與會員列表)。
步驟八:錯誤處理與回傳格式總結
在設計 RESTful API 時,統一且明確的錯誤處理格式,能大幅降低前後端溝通的成本。以下是我們在會員系統中實作的四大錯誤情境與對應的狀態碼:
-
缺少必要欄位 (400 Bad Request)
當前端送來的請求遺漏了必要的參數(如註冊/登入時沒填帳號密碼)。
- 回傳格式:
{"error": "缺少 username 或 password"}
- 回傳格式:
-
帳號或密碼錯誤 (401 Unauthorized)
登入時查無此帳號,或密碼比對失敗。基於資安原則,我們不告知具體是哪一個錯誤。
- 回傳格式: 統一回傳
{"error": "帳號或密碼錯誤"}
- 回傳格式: 統一回傳
-
未登入或 Token 無效 (401 Unauthorized)
呼叫需要授權的 API (如
/api/me或/api/admin/users) 時,未攜帶 Token、Token 過期或比對不到資料。- 回傳格式:
{"error": "未登入或token無效"}
- 回傳格式:
-
權限不足 (403 Forbidden)
使用者已經成功登入(Token 有效),但嘗試存取超過其權限等級的功能(如一般 normal 會員嘗試讀取全站會員列表)。
- 回傳格式:
{"error": "沒有權限,只有admin 可以使用這個功能"}
- 回傳格式: